package com.zhuwei.kafka.consumer;

import com.zhuwei.kafka.constant.Topic;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;


/**
 * kafka消费者消息监听
 * @author zhuiwei
 * @version 2019/0514
 */
@Component
public class KafkaConsumer {

    private final static Logger log=LoggerFactory.getLogger(KafkaConsumer.class);

//    @KafkaListener(topics = Topic.FIRST)
    public void consumer(ConsumerRecord<String,Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer){
        log.info("消费者消费的topic："+topic+";消费的内容：" + record.value() );
        /*
         * 如果需要手工提交异步 consumer.commitSync();
         * 手工同步提交 consumer.commitAsync()
         */
    }
    @KafkaListener(topics = "three")
    public void consumer1(ConsumerRecord<String,Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer){
        log.info("消费者消费的topic："+topic+";消费的内容：" + record.value() );
        /*
         * 如果需要手工提交异步 consumer.commitSync();
         * 手工同步提交 consumer.commitAsync()
         */
    }

    /**
     * consumer2，组group，监听分区0,1
     * 测试：一条消息只能被组中一个消费者消费，且同组中不会重复消费消息
     * 条件：分区数3
     * 结果：kafka群组保证每个分区只能被同一个消费者组的一个消费者使用，但此次同个组中的消费者都消费了分区1 的数据
     */
//    @KafkaListener(id = "consumer2",groupId = "group1", topics = Topic.GROUP,topicPartitions = {@TopicPartition(topic = Topic.GROUP, partitions = {"0", "1"})})
    public void consumer2(ConsumerRecord<String,Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer){
        log.info("消费者consumer2消费的topic："+topic+";消费的内容：" + record.value() );
    }

    /**
     * consumer3，组group，监听分区1,2
     * 测试：一条消息只能被组中一个消费者消费，且同组中不会重复消费消息
     * 结果：kafka机制每个分区只能被同一个消费者组的一个消费者消费，但此测试发现同个组中的消费者都消费了分区1 的数据
     */
//    @KafkaListener(id = "consumer3-1g",groupId = "group1",  topics = Topic.GROUP,topicPartitions = {@TopicPartition(topic = Topic.GROUP, partitions = {"1", "2"})})
    public void consumer3(ConsumerRecord<String,Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer){
        log.info("消费者consumer3消费的topic："+topic+";消费的内容：" + record.value() );
    }

    /**
     * consumer4，组group2，监听所有分区
     * 测试：不同组可以消费相同信息
     * 结果：同kafka机制不同组可以消费相同信息
     */
//    @KafkaListener(id = "consumer4-2g",groupId = "group2",  topics = Topic.GROUP)
    public void consumer4(ConsumerRecord<String,Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer){
        log.info("消费者consumer4消费的topic："+topic+";消费的内容：" + record.value() );
    }
}
